package spring.cloud.tasks.common.zookeeper;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.Getter;
import lombok.NonNull;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import spring.cloud.tasks.client.ItemTaskWorker;
import spring.cloud.tasks.client.Worker;
import spring.cloud.tasks.client.WorkerInfo;
import spring.cloud.tasks.client.item_task_processor.*;
import spring.cloud.tasks.client.item_task_processor.task_item.ClientTaskItemContractExtensionInfo;
import spring.cloud.tasks.client.item_task_processor.task_item.ServerTaskItemContractExtensionInfo;
import spring.cloud.tasks.client.item_task_processor.task_item.ServerTaskItemsSnapshotInfo;
import spring.cloud.tasks.common.utils.Schedules;
import spring.cloud.tasks.common.utils.Uuids;

import java.io.FileNotFoundException;
import java.sql.Timestamp;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * tbschedule回归版 8.15 集中式管理 任务调度执行框架
 */
@Slf4j
public class ZooKeeperItemTaskDataManager extends ZooKeeperDataManager {
    @Setter
    @Getter
    private ZooKeeperManager zooKeeperManager;
    @Setter
    @Getter
    private long zooKeeperBaseDateTimeLong = 0;
    @Setter
    @Getter
    private long localBaseDateTimeLong = 0;
    @Setter
    @Getter
    private String taskRootPath;
    @Setter
    @Getter
    private String workerIdsRootPath;

    public ZooKeeperItemTaskDataManager(ZooKeeperManager zooKeeperManager) throws Exception {
        this.zooKeeperManager = zooKeeperManager;
        //
        this.localBaseDateTimeLong = System.currentTimeMillis();
        String systemTimePath = this.zooKeeperManager.getZooKeeper().create(this.zooKeeperManager.getRootPath() + ZookeeperNodeConstants.$system_time, null, this.zooKeeperManager.getAclList(), CreateMode.EPHEMERAL_SEQUENTIAL);
        this.zooKeeperBaseDateTimeLong = this.zooKeeperManager.getZooKeeper().exists(systemTimePath, false).getCtime();
        ZooKeepers.directoryStructureDelete(getZooKeeper(), systemTimePath);
        if (Math.abs(this.zooKeeperBaseDateTimeLong - this.localBaseDateTimeLong) > 5000) {
            log.error("Zookeeper " + Math.abs(this.zooKeeperBaseDateTimeLong - this.localBaseDateTimeLong) + " ms");
        }
        //
        this.taskRootPath = this.zooKeeperManager.getRootPath() + ZookeeperNodeConstants.ItemTask.$task;
        if (this.getZooKeeper().exists(this.taskRootPath, false) == null) {
            ZooKeepers.createPath(getZooKeeper(), this.taskRootPath, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        }
        this.workerIdsRootPath = this.zooKeeperManager.getRootPath() + ZookeeperNodeConstants.ItemTask.$workerIds;
        if (this.getZooKeeper().exists(this.workerIdsRootPath, false) == null) {
            ZooKeepers.createPath(getZooKeeper(), this.workerIdsRootPath, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        }
    }

    public ZooKeeper getZooKeeper() throws Exception {
        return this.zooKeeperManager.getZooKeeper();
    }

    @Override
    public List<String> registerWorkerToPermittedTaskList(Worker worker) {
        return null;
    }

    @Override
    public List<String> registerWorkerToPermittedTaskListWhenPongPong(Worker worker) {
        return null;
    }

    public long zooKeeperCurrentTimeMillis() {
        return (this.zooKeeperBaseDateTimeLong - this.localBaseDateTimeLong) + System.currentTimeMillis();//求的是zk时间
    }

    public String getTaskPath(String taskId) {
        return this.taskRootPath + ZookeeperNodeConstants.slash + taskId;
    }

    public String getTaskItemsPath(String taskId) {
        return getTaskPath(taskId) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.taskItems;
    }

    public String getTaskItemPath(String taskId, String taskItem) {
        return getTaskItemsPath(taskId) + ZookeeperNodeConstants.slash + taskItem;
    }

    public String getTaskItemsSnapshotInfoPath(String taskId) {
        return getTaskPath(taskId) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.taskItemsSnapshotInfo;
    }

    //
    public String getProcessorsPath(String taskId) {
        return getTaskPath(taskId) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.processorIds;
    }

    public String getProcessorPath(String taskId, String processorId) {
        return getProcessorsPath(taskId) + ZookeeperNodeConstants.slash + processorId;
    }

    public String getWorkerPath(String workerId) {
        return this.workerIdsRootPath + ZookeeperNodeConstants.slash + workerId;
    }

    public String getTaskWorkersPath(String taskId) {
        return this.taskRootPath + ZookeeperNodeConstants.slash + taskId + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.workerIds;
    }

    public String getTaskWorkerPath(String taskId, String workerId) {
        return getTaskWorkersPath(taskId) + ZookeeperNodeConstants.slash + workerId;
    }

    //
    //Task start
    public void createTask(Task task) throws Exception {
        //taskId不能包含$
        if (task.getTaskId().indexOf(ZookeeperNodeConstants.$) > 0) {
            throw new IllegalArgumentException("taskId不能包含$");
        }
        String taskPath = getTaskPath(task.getTaskId());
        String value = JSONObject.toJSONString(task);
        if (this.getZooKeeper().exists(taskPath, false) != null) {
            throw new IllegalArgumentException("task已经存在");
        }
        this.getZooKeeper().create(taskPath, value.getBytes(), this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
    }

    public void pauseTask(String taskId) throws Exception {
        Task task = this.loadTask(taskId);
        task.setStatus(Task.status_pause);
        this.updateTask(task);
    }

    public void runOrRecoverTask(String taskId) throws Exception {
        Task taskType = this.loadTask(taskId);
        taskType.setStatus(Task.status_run);
        this.updateTask(taskType);
    }

    public void updateTask(Task task) throws Exception {
        //taskId不能包含$
        if (task.getTaskId().indexOf(ZookeeperNodeConstants.$) > 0) {
            throw new IllegalArgumentException("taskId不能包含$");
        }
        String taskPath = getTaskPath(task.getTaskId());
        String value = JSONObject.toJSONString(task);
        if (this.getZooKeeper().exists(taskPath, false) == null) {
            this.getZooKeeper().create(taskPath, value.getBytes(), this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        } else {
            this.getZooKeeper().setData(taskPath, value.getBytes(), -1);
        }

    }

    public void cleanTask(String taskId) throws Exception {
        String taskPath = getTaskPath(taskId);
        ZooKeepers.directoryStructureDelete(this.getZooKeeper(), taskPath);
    }

    public List<Task> getTaskList() throws Exception {
        List<Task> taskList = new ArrayList<Task>();
        List<String> taskIdList = this.getZooKeeper().getChildren(this.taskRootPath, false);
        Collections.sort(taskIdList);
        for (String taskId : taskIdList) {
            taskList.add(this.loadTask(taskId));
        }
        return taskList;
    }

    public Task loadTask(String taskId) throws Exception {
        String taskPath = getTaskPath(taskId);
        if (this.getZooKeeper().exists(taskPath, false) == null) {
            return null;
        }
        String value = new String(this.getZooKeeper().getData(taskPath, false, null));
        Task task = JSONObject.parseObject(value, Task.class);
        return task;
    }

    //Task end
    //Task Item start
    public void createTaskItems(String taskId) throws Exception {
        String taskItemsPath = getTaskItemsPath(taskId);
        try {
            ZooKeepers.directoryStructureDelete(this.getZooKeeper(), taskItemsPath);
        } catch (Exception e) {
            if (e instanceof KeeperException
                    && ((KeeperException) e).code().intValue() == KeeperException.Code.SESSIONEXPIRED.intValue()) {
                log.warn("delete : zookeeper sessionzookeeper");
                zooKeeperManager.reConnect();
                ZooKeepers.directoryStructureDelete(this.getZooKeeper(), taskItemsPath);
            }
        }
        Task task = this.loadTask(taskId);
        TaskItemsInfo taskItemInfo = new TaskItemsInfo();
        task.setTaskItemsVersion(task.getTaskItemsVersion() + 1);
        updateTask(task);//失败了也没有什么
        taskItemInfo.setVersion(task.getTaskItemsVersion());
        taskItemInfo.setStatus(task_items_status_initing);
        this.getZooKeeper().create(taskItemsPath, JSON.toJSONString(taskItemInfo).getBytes(), this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        //
        this.createTaskItems(taskId, task.getTaskParameter(), task.getTaskItems());//分配中
        //
        taskItemInfo.setStatus(task_items_status_inited);
        this.getZooKeeper().setData(taskItemsPath, JSON.toJSONString(taskItemInfo).getBytes(), -1);
    }

    public TaskItem[] createTaskItems(String taskId, String taskParameter, String[] taskItemStrings) throws Exception {
        TaskItem[] taskItems = new TaskItem[taskItemStrings.length];
        for (int i = 0; i < taskItemStrings.length; i++) {
            taskItems[i] = new TaskItem();
            taskItems[i].setTaskId(taskId);
            taskItems[i].setTaskItemString(taskParameter);
            taskItems[i].setTaskItemString(taskItemStrings[i]);

        }

        for (TaskItem taskItem : taskItems) {
            String taskItemsPath = getTaskItemsPath(taskId);
            if (this.getZooKeeper().exists(taskItemsPath, false) == null) {
                ZooKeepers.createPath(this.getZooKeeper(), taskItemsPath, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
            }
            String taskItemPath = getTaskItemPath(taskId, taskItem.getTaskItemString());
            this.getZooKeeper().create(taskItemPath, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
            //

            this.getZooKeeper().create(taskItemPath + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$message, taskItem.getMark().getBytes(), this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
            this.getZooKeeper().create(taskItemPath + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$processor_id, TaskItem.NO_PROCESSOR_ID.getBytes(), this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
            ServerTaskItemContractExtensionInfo serverTaskItemContractExtensionInfo = new ServerTaskItemContractExtensionInfo();
            this.getZooKeeper().create(taskItemPath + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$current_processor_contract_extension_info, (JSON.toJSONString(serverTaskItemContractExtensionInfo)).getBytes(), this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);

        }
        return taskItems;
    }

    //如果任务没有分配则需要等,如果任务分配了,调度器需要立即结束任务的执行,关闭自己。
    public boolean taskItemsIsIniting(String taskId) throws Exception {
        TaskItemsInfo taskItemInfo = getTaskItemInfo(taskId);
        return taskItemInfo != null && task_items_status_initing.equals(taskItemInfo.getStatus());
    }

    //可以拉取任务
    public boolean taskItemsIsInited(String taskId) throws Exception {
        TaskItemsInfo taskItemInfo = getTaskItemInfo(taskId);
        return taskItemInfo != null && task_items_status_inited.equals(taskItemInfo.getStatus());
    }

    public static final String task_items_status_initing = "initing";
    public static final String task_items_status_inited = "inited";

    public TaskItemsInfo getTaskItemInfo(String taskId) throws Exception {
        String taskItemsPath = getTaskItemsPath(taskId);
        if (this.getZooKeeper().exists(taskItemsPath, false) != null) {
            byte[] bytes = this.getZooKeeper().getData(taskItemsPath, false, null);
            if (bytes == null) {
                throw new NullPointerException();
            }
            String value = new String(bytes);
            TaskItemsInfo taskItemInfo = JSONObject.parseObject(value, TaskItemsInfo.class);
            return taskItemInfo;
        }
        return null;
    }

    public List<String> getTaskItemAndParameterList(TaskProcessorInfo taskProcessorInfo) throws Exception {
        String taskId = taskProcessorInfo.getTaskId();
        String processorId = taskProcessorInfo.getProcessorId();
        String taskItemsPath = getTaskItemsPath(taskId);
        List<String> taskItemList = this.getZooKeeper().getChildren(taskItemsPath, false);
        Collections.sort(taskItemList, new TaskItemComparator());
        List<String> taskItemStringList = new ArrayList<String>();
        for (String taskItem : taskItemList) {
            byte[] processorIdBytes = this.getZooKeeper().getData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$processor_id, false, null);
            if (processorId.equals(new String(processorIdBytes))) {
                //需要前面续期
                byte[] currentProcessorIdBytes = this.getZooKeeper().getData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$current_processor_contract_extension_info, false, null);
                ServerTaskItemContractExtensionInfo serverTaskItemContractExtensionInfo = ServerTaskItemContractExtensionInfo.parse(currentProcessorIdBytes);
                if (serverTaskItemContractExtensionInfo.getProcessorId().equals(processorId) && !serverTaskItemContractExtensionInfo.isOverdue()) {
                    taskItemStringList.add(taskItem);
                }

            }
        }
        return taskItemStringList;
    }


    public List<TaskItem> loadTaskItemList(String taskId) throws Exception {
        List<TaskItem> taskItemList = new ArrayList<TaskItem>();
        String taskItemsPath = getTaskItemsPath(taskId);
        if (this.getZooKeeper().exists(taskItemsPath, false) == null) {
            return taskItemList;
        }
        List<String> taskItemStringList = this.getZooKeeper().getChildren(taskItemsPath, false);
        Collections.sort(taskItemStringList, new TaskItemComparator());
        for (String taskItemString : taskItemStringList) {
            String taskItemPath = getTaskItemPath(taskId, taskItemString);
            //
            TaskItem taskItem = new TaskItem();
            taskItem.setTaskId(taskId);
            taskItem.setTaskItemString(taskItemString);

            byte[] dealInfoBytes = this.getZooKeeper().getData(taskItemPath + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$message, false, null);
            taskItem.setMark(new String(dealInfoBytes));
            byte[] processorIdBytes = this.getZooKeeper().getData(taskItemPath + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$processor_id, false, null);
            taskItem.setProcessorId(new String(processorIdBytes));
            byte[] currentProcessorIdBytes = this.getZooKeeper().getData(taskItemPath + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$current_processor_contract_extension_info, false, null);
            ServerTaskItemContractExtensionInfo serverTaskItemContractExtensionInfo = ServerTaskItemContractExtensionInfo.parse(currentProcessorIdBytes);
            taskItem.setServerTaskItemContractExtensionInfo(serverTaskItemContractExtensionInfo);
            if (serverTaskItemContractExtensionInfo.getProcessorId() == null) {
                taskItem.setTaskItemString("暂无续期");
            } else {
                if (serverTaskItemContractExtensionInfo.isOverdue()) {
                    taskItem.setTaskItemString("续期已经过期");
                } else {
                    taskItem.setTaskItemString("正常续期中");
                }
            }
            taskItemList.add(taskItem);
        }
        return taskItemList;

    }
    //Task Item end

    /**
     * Task Item-processor start
     */
    public void getContractExtensionInfos(String taskId, TaskProcessorInfo taskProcessorInfo, ConcurrentHashMap<String, ClientTaskItemContractExtensionInfo> taskItemToClientTaskItemContractExtensionInfoMap) throws Exception {
        String taskItemsPath = getTaskItemsPath(taskId);
        List<String> taskItemList = this.getZooKeeper().getChildren(taskItemsPath, false);

        Collections.sort(taskItemList, new TaskItemComparator());
        //没有调度器进行任务处理的时候 直接清空

        for (int i = 0; i < taskItemList.size(); i++) {
            String taskItem = taskItemList.get(i);
            byte[] processorIdBytes = this.getZooKeeper().getData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$processor_id, false, null);
            String processorId = new String(processorIdBytes);
            //
            //
            byte[] currentProcessorContractExtensionInfoBytes = this.getZooKeeper().getData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$current_processor_contract_extension_info, false, null);
            String currentProcessorContractExtensionInfo = new String(currentProcessorContractExtensionInfoBytes);
            ServerTaskItemContractExtensionInfo serverTaskItemContractExtensionInfo = JSON.parseObject(currentProcessorContractExtensionInfo, ServerTaskItemContractExtensionInfo.class);
            //
            if (taskProcessorInfo.getProcessorId().equals(processorId)) {
                if (serverTaskItemContractExtensionInfo.getProcessorId() == null ||
                        serverTaskItemContractExtensionInfo.getProcessorId().equals(taskProcessorInfo.getProcessorId()) ||
                        !serverTaskItemContractExtensionInfo.getProcessorId().equals(taskProcessorInfo.getProcessorId()) && serverTaskItemContractExtensionInfo.getLastTimeExtensionDate().before(new Date())//直接清理
                ) {
                    serverTaskItemContractExtensionInfo.setProcessorId(taskProcessorInfo.processorId);
                    serverTaskItemContractExtensionInfo.setLastTimeExtensionDate(new Date());//add ...
                    this.getZooKeeper().setData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$current_processor_contract_extension_info, JSON.toJSONString(serverTaskItemContractExtensionInfo).getBytes(), -1);
                    //
                    taskItemToClientTaskItemContractExtensionInfoMap.put(taskItem, new ClientTaskItemContractExtensionInfo());
                    ClientTaskItemContractExtensionInfo clientTaskItemContractExtensionInfo = taskItemToClientTaskItemContractExtensionInfoMap.get(taskItem);
                    clientTaskItemContractExtensionInfo.setTaskItem(taskItem);
                    clientTaskItemContractExtensionInfo.setStatus("Y");
                    clientTaskItemContractExtensionInfo.setLastTimeExtensionDate(new Date());
                    clientTaskItemContractExtensionInfo.setLastTimeExtensionDateInProsessor(new Date());
                    continue;
                }
            }
        }

    }

    public void updateContractExtensionInfos(String taskId, String taskItem, TaskProcessorInfo taskProcessorInfo, ConcurrentHashMap<String, ClientTaskItemContractExtensionInfo> taskItemToClientTaskItemContractExtensionInfoMap) throws Exception {
        byte[] processorIdBytes = this.getZooKeeper().getData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$processor_id, false, null);
        String processorId = new String(processorIdBytes);
        //
        //
        byte[] currentProcessorContractExtensionInfoBytes = this.getZooKeeper().getData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$current_processor_contract_extension_info, false, null);
        String currentProcessorContractExtensionInfo = new String(currentProcessorContractExtensionInfoBytes);
        ServerTaskItemContractExtensionInfo serverTaskItemContractExtensionInfo = JSON.parseObject(currentProcessorContractExtensionInfo, ServerTaskItemContractExtensionInfo.class);
        //
        if (taskProcessorInfo.getProcessorId().equals(processorId)) {//如果长时间没有续期 那么这个会被清空
            if (serverTaskItemContractExtensionInfo.getProcessorId() == null ||
                    serverTaskItemContractExtensionInfo.getProcessorId().equals(taskProcessorInfo.getProcessorId()) ||
                    !serverTaskItemContractExtensionInfo.getProcessorId().equals(taskProcessorInfo.getProcessorId()) && serverTaskItemContractExtensionInfo.getLastTimeExtensionDate().before(new Date())//这种情况可能瞬间出现
            ) {
                serverTaskItemContractExtensionInfo.setProcessorId(taskProcessorInfo.processorId);
                serverTaskItemContractExtensionInfo.setLastTimeExtensionDate(new Date());//add ...
                this.getZooKeeper().setData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$current_processor_contract_extension_info, JSON.toJSONString(serverTaskItemContractExtensionInfo).getBytes(), -1);
                //
                taskItemToClientTaskItemContractExtensionInfoMap.put(taskItem, new ClientTaskItemContractExtensionInfo());
                ClientTaskItemContractExtensionInfo clientTaskItemContractExtensionInfo = taskItemToClientTaskItemContractExtensionInfoMap.get(taskItem);
                clientTaskItemContractExtensionInfo.setTaskItem(taskItem);
                clientTaskItemContractExtensionInfo.setStatus("Y");
                clientTaskItemContractExtensionInfo.setLastTimeExtensionDate(new Date());
                clientTaskItemContractExtensionInfo.setLastTimeExtensionDateInProsessor(new Date());
                return;
            }
        } else {
            taskItemToClientTaskItemContractExtensionInfoMap.remove(taskItem);
        }
    }

    /**
     * ProcessorId如果为当前taskItem处理者,可以将request_Processor_id提升为处理者
     */
    public void releaseContractExtensionInfos(TaskProcessorInfo taskProcessorInfo, ConcurrentHashMap<String, ClientTaskItemContractExtensionInfo> taskItemToClientTaskItemContractExtensionInfoMap) throws Exception {
        String taskId = taskProcessorInfo.getTaskId();
        String processorId = taskProcessorInfo.getProcessorId();

        String taskItemsPath = getTaskItemsPath(taskId);
        Set<String> taskItemSet = new HashSet<String>();

        for (String taskItem : this.getZooKeeper().getChildren(taskItemsPath, false)) {

            byte[] currentProcessorContractExtensionInfoBytes = this.getZooKeeper().getData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$current_processor_contract_extension_info, false, null);
            String currentProcessorContractExtensionInfo = new String(currentProcessorContractExtensionInfoBytes);
            ServerTaskItemContractExtensionInfo serverTaskItemContractExtensionInfo = JSON.parseObject(currentProcessorContractExtensionInfo, ServerTaskItemContractExtensionInfo.class);

            if (serverTaskItemContractExtensionInfo.getProcessorId() != null && serverTaskItemContractExtensionInfo.getProcessorId().equals(processorId)) {
                taskItemSet.add(serverTaskItemContractExtensionInfo.getProcessorId());
            }
        }
        releaseContractExtensionInfos(taskId, new ArrayList(taskItemSet), taskProcessorInfo, taskItemToClientTaskItemContractExtensionInfoMap);
        this.updateTaskProcessorsReloadFlag(taskId);
    }

    public void releaseContractExtensionInfos(String taskId, List<String> taskItemList, TaskProcessorInfo taskProcessorInfo, ConcurrentHashMap<String, ClientTaskItemContractExtensionInfo> taskItemToClientTaskItemContractExtensionInfoMap) throws Exception {
        for (String taskItem : taskItemList) {
            releaseContractExtensionInfos(taskId, taskItem, taskProcessorInfo, taskItemToClientTaskItemContractExtensionInfoMap);
        }
    }

    public void releaseContractExtensionInfos(String taskId, String taskItem, TaskProcessorInfo taskProcessorInfo, ConcurrentHashMap<String, ClientTaskItemContractExtensionInfo> taskItemToClientTaskItemContractExtensionInfoMap) throws Exception {
        byte[] currentProcessorContractExtensionInfoBytes = this.getZooKeeper().getData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$current_processor_contract_extension_info, false, null);
        String currentProcessorContractExtensionInfo = new String(currentProcessorContractExtensionInfoBytes);
        ServerTaskItemContractExtensionInfo serverTaskItemContractExtensionInfo = JSON.parseObject(currentProcessorContractExtensionInfo, ServerTaskItemContractExtensionInfo.class);
        //
        if (taskProcessorInfo.getProcessorId().equals(serverTaskItemContractExtensionInfo.getProcessorId())) {
            serverTaskItemContractExtensionInfo.setProcessorId(null);
            serverTaskItemContractExtensionInfo.setLastTimeExtensionDate(new Date());//add ...
            this.getZooKeeper().setData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$current_processor_contract_extension_info, JSON.toJSONString(serverTaskItemContractExtensionInfo).getBytes(), -1);
        }
        taskItemToClientTaskItemContractExtensionInfoMap.remove(taskItem);

    }
    //

    /**
     * 没有必要进行额外的状态检测 只需要分配任务的时候进行处理失效的task item info with processor
     *
     * @param taskId
     * @param processorIdList
     * @throws Exception
     */
    public void assignTaskItemWithProcessorId(String taskId, List<String> processorIdList) throws Exception {

        String taskItemsPath = getTaskItemsPath(taskId);
        List<String> taskItemList = this.getZooKeeper().getChildren(taskItemsPath, false);

        Collections.sort(taskItemList, new TaskItemComparator());

        ServerTaskItemsSnapshotInfo newTaskItemsSnapshotInfo = new ServerTaskItemsSnapshotInfo();
        String taskItemsSnapshotInfoPath = getTaskItemsSnapshotInfoPath(taskId);
        byte[] taskItemsSnapshotInfoBytes = this.getZooKeeper().getData(taskItemsSnapshotInfoPath, false, null);//需要初始化一个空的对象
        ServerTaskItemsSnapshotInfo oldtaskItemsSnapshotInfo = ServerTaskItemsSnapshotInfo.parse(taskItemsSnapshotInfoBytes);

        if (processorIdList.size() <= 0) {
            for (int i = 0; i < taskItemList.size(); i++) {
                String taskItem = taskItemList.get(i);
                this.getZooKeeper().setData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$processor_id, TaskItem.NO_PROCESSOR_ID.getBytes(), -1);
                ServerTaskItemContractExtensionInfo serverTaskItemContractExtensionInfo = new ServerTaskItemContractExtensionInfo();
                this.getZooKeeper().setData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$current_processor_contract_extension_info, (JSON.toJSONString(serverTaskItemContractExtensionInfo)).getBytes(), -1);
            }
            for (int i = 0; i < taskItemList.size(); i++) {
                String taskItem = taskItemList.get(i);
                newTaskItemsSnapshotInfo.taskItemToProcessorIdMap.put(taskItem, TaskItem.NO_PROCESSOR_ID);
            }
            this.getZooKeeper().setData(taskItemsSnapshotInfoPath, oldtaskItemsSnapshotInfo.getBytes(), -1);
            this.updateTaskProcessorsReloadFlag(taskId);
            return;
        }
        //

        //
        for (int i = 0; i < taskItemList.size(); i++) {
            String taskItem = taskItemList.get(i);
            //
            byte[] currentProcessorContractExtensionInfoBytes = this.getZooKeeper().getData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$current_processor_contract_extension_info, false, null);
            ServerTaskItemContractExtensionInfo serverTaskItemContractExtensionInfo = ServerTaskItemContractExtensionInfo.parse(currentProcessorContractExtensionInfoBytes);
            //
            if (serverTaskItemContractExtensionInfo.getProcessorId() != null && serverTaskItemContractExtensionInfo.isOverdue()) {
                //没有进行续期
                serverTaskItemContractExtensionInfo.setProcessorId(null);
                this.getZooKeeper().setData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$current_processor_contract_extension_info, serverTaskItemContractExtensionInfo.getBytes(), -1);
                continue;
            }
        }
        List<String> oldProcessorIdList = new ArrayList<String>();
        for (int i = 0; i < taskItemList.size(); i++) {
            String taskItem = taskItemList.get(i);
            byte[] processorIdBytes = this.getZooKeeper().getData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$processor_id, false, null);
            String processorId = new String(processorIdBytes);
            oldProcessorIdList.add(processorId);
        }
        if (processorIdList.containsAll(oldProcessorIdList) && oldProcessorIdList.containsAll(processorIdList)) {
            return;//处理器没有改变维持原来的-如果存在变化:新的处理器数增加或者减少。或者处理的服务器增加或者减少。
        }
        Map<String, String> taskItemToProcessorIdMap = Schedules.allot(taskItemList, processorIdList);//taskItemList size/processorId num
        newTaskItemsSnapshotInfo.taskItemToProcessorIdMap.putAll(taskItemToProcessorIdMap);
        for (int i = 0; i < taskItemList.size(); i++) {
            String taskItem = taskItemList.get(i);
            this.getZooKeeper().setData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$processor_id, taskItemToProcessorIdMap.get(taskItem).getBytes(), -1);
            ServerTaskItemContractExtensionInfo serverTaskItemContractExtensionInfo = new ServerTaskItemContractExtensionInfo();
            this.getZooKeeper().setData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$current_processor_contract_extension_info, (JSON.toJSONString(serverTaskItemContractExtensionInfo)).getBytes(), -1);
        }
        this.updateTaskProcessorsReloadFlag(taskId);//重新分配以后 那么容许通知客户端，以及让客户端一段时间将自己处理的任务停掉。
    }


    public long updateTaskProcessorsReloadFlag(String taskId) throws Exception {
        String processorsPath = getProcessorsPath(taskId);
        Stat stat = this.getZooKeeper().setData(processorsPath, "reload=true".getBytes(), -1);
        return stat.getVersion();

    }

    public void cleanTaskItemCurrentProcessorIdOutOfProcessorIdList(String taskId, List<String> processorIdList) throws Exception {
        String taskItemsPath = getTaskItemsPath(taskId);
        for (String taskItem : this.getZooKeeper().getChildren(taskItemsPath, false)) {
            //优先处理processorId
            byte[] processorIdBytes = this.getZooKeeper().getData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$processor_id, false, null);
            String processorId = new String(processorIdBytes);
            boolean isFind = false;
            for (String $processorId : processorIdList) {
                if ($processorId.equals(processorId)) {
                    isFind = true;
                    break;
                }
            }
            //找不到则清空
            if (isFind == false) {
                this.getZooKeeper().setData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$processor_id, TaskItem.NO_PROCESSOR_ID.getBytes(), -1);
            }
            //
            byte[] current_processor_contract_extension_infoBytes = this.getZooKeeper().getData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$current_processor_contract_extension_info, false, null);
            ServerTaskItemContractExtensionInfo serverTaskItemContractExtensionInfo = ServerTaskItemContractExtensionInfo.parse(current_processor_contract_extension_infoBytes);
            isFind = false;
            for (String $processorId : processorIdList) {
                if ($processorId.equals(serverTaskItemContractExtensionInfo.getProcessorId())) {
                    isFind = true;
                    break;
                }
            }
            //找不到则清空
            if (isFind == false) {
                serverTaskItemContractExtensionInfo = new ServerTaskItemContractExtensionInfo();
                this.getZooKeeper().setData(getTaskItemPath(taskId, taskItem) + ZookeeperNodeConstants.slash + ZookeeperNodeConstants.ItemTask.TaskItem.$current_processor_contract_extension_info, serverTaskItemContractExtensionInfo.getBytes(), -1);

            }

        }

    }


    /**
     * Task Item-processor end
     */
    //processor start
    public void registerProcessor(@NonNull TaskProcessorInfo taskProcessorInfo) throws Exception {
        if (taskProcessorInfo.isRegister() == true) {
            throw new Exception(taskProcessorInfo.getProcessorId() + " ");
        }
        String taskPath = getTaskPath(taskProcessorInfo.getTaskId());
        if (this.getZooKeeper().exists(taskPath, false) == null) {
            this.getZooKeeper().create(taskPath, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        }
        //
        String processorsPath = getProcessorsPath(taskProcessorInfo.getTaskId());
        if (this.getZooKeeper().exists(processorsPath, false) == null) {
            this.getZooKeeper().create(processorsPath, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        }
        //
        String processorId = taskProcessorInfo.getIp() + ZookeeperNodeConstants.$ + Uuids.getUuid();
        String processorPath = getProcessorPath(taskProcessorInfo.getTaskId(), processorId);
        Timestamp heartBeatTime = new Timestamp(this.zooKeeperCurrentTimeMillis());
        taskProcessorInfo.setHeartBeatTime(heartBeatTime);
        String value = JSONObject.toJSONString(taskProcessorInfo);
        this.getZooKeeper().create(processorPath, value.getBytes(), this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT_SEQUENTIAL);
        //
        taskProcessorInfo.setProcessorId(processorId);
        taskProcessorInfo.setRegister(true);
    }

    public TaskProcessorInfo getProcessorInfo(String taskId, String processorId) throws Exception {
        String processorPath = getProcessorPath(taskId, processorId);
        String value = new String(this.getZooKeeper().getData(processorPath, false, null));
        TaskProcessorInfo taskProcessorInfo = JSONObject.parseObject(value, TaskProcessorInfo.class);
        return taskProcessorInfo;
    }

    /**
     * @param taskProcessorInfo
     * @return true 存在,false 不存在,如果抛出异常则可能是网络的原因
     * @throws Exception
     */
    public boolean updateProcessorHeartBeatTimeAndVersion(TaskProcessorInfo taskProcessorInfo) throws Exception {
        Timestamp heartBeatTime = new Timestamp(this.zooKeeperCurrentTimeMillis());
        String processorPath = getProcessorPath(taskProcessorInfo.getTaskId(), taskProcessorInfo.getProcessorId());
        //节点不存在则直接返回
        if (this.getZooKeeper().exists(processorPath, false) == null) {
            taskProcessorInfo.setRegister(false);
            return false;
        } else {
            Date oldHeartBeatTime = taskProcessorInfo.getHeartBeatTime();
            taskProcessorInfo.setHeartBeatTime(heartBeatTime);
            taskProcessorInfo.setVersion(taskProcessorInfo.getVersion() + 1);
            String value = JSONObject.toJSONString(taskProcessorInfo);
            try {
                this.getZooKeeper().setData(processorPath, value.getBytes(), -1);
                return true;
            } catch (Exception e) {
                taskProcessorInfo.setHeartBeatTime(oldHeartBeatTime);
                taskProcessorInfo.setVersion(taskProcessorInfo.getVersion() - 1);//为什么-1,这样用来恢复原来的记录
                throw e;
            }
        }
    }


    public void unRegisterProcessor(TaskProcessorInfo taskProcessorInfo) throws Exception {
        String taskId = taskProcessorInfo.getTaskId();
        String processorId = taskProcessorInfo.getProcessorId();
        String processorPath = getProcessorPath(taskId, processorId);
        if (this.getZooKeeper().exists(processorPath, false) != null) {
            this.getZooKeeper().delete(processorPath, -1);
        }
    }

    public int clearExpireProcessors(String taskId, long expireTime) throws Exception {
        int count = 0;
        String processorsPath = getProcessorsPath(taskId);
        if (this.getZooKeeper().exists(processorsPath, false) == null) {
            String taskPath = getTaskPath(taskId);// ../
            if (this.getZooKeeper().exists(taskPath, false) == null) {
                this.getZooKeeper().create(taskPath, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
            }
            //
            this.getZooKeeper().create(processorsPath, null, this.zooKeeperManager.getAclList(), CreateMode.PERSISTENT);
        }
        for (String processorId : this.getZooKeeper().getChildren(processorsPath, false)) {
            try {
                String processorPath = getProcessorPath(taskId, processorId);
                Stat stat = this.getZooKeeper().exists(processorPath, false);
                if (zooKeeperCurrentTimeMillis() - stat.getMtime() > expireTime) {//删除过期的server
                    ZooKeepers.directoryStructureDelete(this.getZooKeeper(), processorPath);
                    count++;
                }
            } catch (Exception e) {

                count++;
            }
        }
        return count;
    }

    public List<TaskProcessorInfo> getTaskProcessorInfoList(String taskId)
            throws Exception {
        List<TaskProcessorInfo> taskProcessorInfoList = new ArrayList<TaskProcessorInfo>();

        String processorsPath = getProcessorsPath(taskId);
        if (this.getZooKeeper().exists(processorsPath, false) == null) {
            return taskProcessorInfoList;
        }
        List<String> processorIdList = this.getZooKeeper().getChildren(processorsPath, false);
        Collections.sort(processorIdList, new ProcessorComparator());
        for (String processorId : processorIdList) {
            try {
                String processorPath = getProcessorPath(taskId, processorId);
                String string = new String(this.getZooKeeper().getData(processorPath, false, null));
                TaskProcessorInfo taskTaskProcessorInfo = JSONObject.parseObject(string, TaskProcessorInfo.class);
                taskTaskProcessorInfo.setCenterServerTime(new Timestamp(this.zooKeeperCurrentTimeMillis()));
                taskProcessorInfoList.add(taskTaskProcessorInfo);
            } catch (Exception e) {
                log.debug(e.getMessage(), e);
            }
        }
        return taskProcessorInfoList;
    }

    public List<TaskProcessorInfo> getTaskProcessorInfoListByWorkerId(String workerId) throws Exception {
        List<TaskProcessorInfo> taskProcessorInfoList = new ArrayList<TaskProcessorInfo>();
        for (String taskId : this.getZooKeeper().getChildren(this.taskRootPath, false)) {
            String processorsPath = getProcessorsPath(taskId);
            for (String processorId : this.getZooKeeper().getChildren(processorsPath, false)) {
                String processorPath = getProcessorPath(taskId, processorId);
                String string = new String(this.getZooKeeper().getData(processorPath, false, null));
                TaskProcessorInfo taskTaskProcessorInfo = JSONObject.parseObject(string, TaskProcessorInfo.class);
                taskTaskProcessorInfo.setCenterServerTime(new Timestamp(this.zooKeeperCurrentTimeMillis()));
                if (taskTaskProcessorInfo.getWorkerId().equals(workerId))//全部遍历 找到machineId相关的task handler
                {
                    taskProcessorInfoList.add(taskTaskProcessorInfo);
                }
            }
        }
        Collections.sort(taskProcessorInfoList, new ProcessorComparator2());
        return taskProcessorInfoList;
    }

    public List<TaskProcessorInfo> getProcessorInfoList(
            String taskId,//
            String ip, //
            String orderString) throws Exception {//
        List<TaskProcessorInfo> taskTaskProcessorInfoList = new ArrayList<TaskProcessorInfo>();
        if (ip == null) {
            taskTaskProcessorInfoList.addAll(this.getTaskProcessorInfoList(taskId));
        } else {
            for (TaskProcessorInfo taskTaskProcessorInfo : this.getTaskProcessorInfoList(taskId)) {
                if (ip.equals(taskTaskProcessorInfo.getIp()))//ip限制
                {
                    taskTaskProcessorInfoList.add(taskTaskProcessorInfo);
                }
            }
        }
        Collections.sort(taskTaskProcessorInfoList, new TaskProcessorInfoComparator(orderString));

        return taskTaskProcessorInfoList;
    }

    public long getProcessorsStatVersion(String taskId) throws Exception {
        String processorsPath = getProcessorsPath(taskId);
        Stat stat = new Stat();
        this.getZooKeeper().getData(processorsPath, false, stat);
        return stat.getVersion();
    }


    public List<String> loadProcessorIdList(String taskId) throws Exception {
        String processorsPath = getProcessorsPath(taskId);
        if (this.getZooKeeper().exists(processorsPath, false) == null) {
            return new ArrayList<String>();
        }
        List<String> processorIdList = this.getZooKeeper().getChildren(processorsPath, false);
        Collections.sort(processorIdList, new Comparator<String>() {//
            public int compare(String left, String right) {//
                return left.substring(left.lastIndexOf("$") + 1)//
                        .compareTo(//
                                right.substring(right.lastIndexOf("$") + 1));//
            }
        });
        return processorIdList;
    }
    //processor end

    //itemTaskWorker start
    public WorkerInfo loadWorkerInfo(String workerId) throws Exception {
        String workerPath = getWorkerPath(workerId);
        if (this.getZooKeeper().exists(workerPath, false) == null) {
            throw new FileNotFoundException(workerId);
        }
        byte[] bytes = this.getZooKeeper().getData(workerPath, false, null);
        if (bytes == null) {
            throw new FileNotFoundException();
        }
        WorkerInfo workerInfo = JSON.parseObject(new String(bytes), WorkerInfo.class);
        return workerInfo;
    }

    //
    public List<WorkerInfo> getWorkerInfoList() throws Exception {

        List<WorkerInfo> workerInfoList = new ArrayList<WorkerInfo>();
        List<String> workertIdList = this.getZooKeeper().getChildren(this.workerIdsRootPath, false);
        Collections.sort(workertIdList, new Comparator<String>() {//
            public int compare(String leftString, String rightString) {//
                return leftString.substring(leftString.lastIndexOf(ZookeeperNodeConstants.$) + 1)//
                        .compareTo(//
                                rightString.substring(rightString.lastIndexOf(ZookeeperNodeConstants.$) + 1));//
            }
        });
        for (String workertId : workertIdList) {
            String workerPath = getWorkerPath(workertId);
            byte[] bytes = this.getZooKeeper().getData(workerPath, false, null);
            if (bytes == null) {
                throw new NullPointerException();
            } else {
                WorkerInfo workerInfo = JSON.parseObject(new String(bytes), WorkerInfo.class);
                workerInfoList.add(workerInfo);
            }
        }
        return workerInfoList;
    }

    /**
     * 上线和下线处理
     */
    public void updateWorkerInfo(String workertId, boolean executeTasks) throws Exception {
        String workerPath = getWorkerPath(workertId);
        if (this.getZooKeeper().exists(workerPath, false) == null) {
            throw new Exception(workertId);
        }
        byte[] bytes = this.getZooKeeper().getData(workerPath, false, null);
        if (bytes == null) {
            throw new NullPointerException();
        } else {
            WorkerInfo workerInfo = JSON.parseObject(new String(bytes), WorkerInfo.class);
            workerInfo.setExecuteTasks(executeTasks);
            this.getZooKeeper().setData(workerPath, JSON.toJSONString(workerInfo).getBytes(), -1);
        }

    }

    public List<String> registerWorkerToPermittedTaskList(ItemTaskWorker worker) throws Exception {
        //String workerId  ip+itemTaskWorker.getHostName()+uuid
        String workerId = worker.getIp() + ZookeeperNodeConstants.$ + Uuids.getUuid();
        String workerPath = getWorkerPath(workerId);
        //
        WorkerInfo workerInfo = new WorkerInfo();
        workerInfo.setWorkerId(workerId);
        workerInfo.setExecuteTasks(true);//立即可以执行
        workerInfo.setTaskIdList(worker.getTaskIdList());
        this.getZooKeeper().create(workerPath, JSON.toJSONString(workerInfo).getBytes(), this.zooKeeperManager.getAclList(), CreateMode.EPHEMERAL_SEQUENTIAL);
        //zk注册成功后更新本地
        worker.setWorkerId(workerId);
        //
        List<String> taskIdList = getTaskIdList(worker);
        return taskIdList;
    }

    public List<String> registerWorkerToPermittedTaskListWhenPongPong(ItemTaskWorker worker) throws Exception {

        String workerPath = getWorkerPath(worker.getWorkerId());
        if (this.getZooKeeper().exists(workerPath, false) != null) {
            byte[] bytes = this.getZooKeeper().getData(workerPath, false, null);
            if (bytes == null) {
                throw new NullPointerException();
            } else {
                WorkerInfo workerInfo = JSON.parseObject(new String(bytes), WorkerInfo.class);
                boolean executeTasks = workerInfo.isExecuteTasks();
                if (executeTasks) {
                    if (!worker.executeTasks) {//之前任务停止 后来允许开启
                        worker.executeTasks = true;
                    }
                    List<String> taskIdList = getTaskIdList(worker);
                    return taskIdList;

                } else {
                    if (worker.executeTasks) {
                        worker.executeTasks = false;
                    }
                    List<String> taskIdList = new ArrayList<String>();
                    return taskIdList;
                }
            }
        } else {
            return registerWorkerToPermittedTaskList(worker);//完完全全重新注册
        }
    }

    private List<String> getTaskIdList(ItemTaskWorker worker) throws Exception {
        List<String> taskIdList = new ArrayList<String>();
        for (Task task : getTaskList()) {//
            boolean isFind = false;//
            if (worker.getTaskIdList().contains(task.getTaskId()) &&//task支持判断
                    !Task.status_pause.equalsIgnoreCase(task.getStatus()) &&//非暂停状态
                    (task.getIpList() == null || (task.getIpList() != null && task.getIpList().contains(worker.getIp())))) {//IP限制
                String taskIdAndWorkerIdPath = getTaskWorkerPath(task.getTaskId(), worker.getWorkerId());
                if (this.getZooKeeper().exists(taskIdAndWorkerIdPath, false) == null) {
                    this.getZooKeeper().create(taskIdAndWorkerIdPath, null, this.zooKeeperManager.getAclList(), CreateMode.EPHEMERAL);
                }
                isFind = true;
                break;
            }
            //删除不必存在的数据
            if (!isFind) {
                String taskIdAndWorkerIdPath = getTaskWorkerPath(task.getTaskId(), worker.getWorkerId());
                if (this.getZooKeeper().exists(taskIdAndWorkerIdPath, false) != null) {
                    ZooKeepers.directoryStructureDelete(this.getZooKeeper(), taskIdAndWorkerIdPath);
                    taskIdList.add(task.getTaskId());
                }
            }
        }
        return taskIdList;
    }


    /**
     * 卸载
     */
    public void unRegisterWorker(Worker worker) throws Exception {
        for (String taskId : this.getZooKeeper().getChildren(this.taskRootPath, false)) {
            String taskIdAndWorkerIdPath = getTaskWorkerPath(taskId, worker.getWorkerId());
            if (this.getZooKeeper().exists(taskIdAndWorkerIdPath, false) != null) {
                ZooKeepers.directoryStructureDelete(this.getZooKeeper(), taskIdAndWorkerIdPath);
            }
        }
    }


    public List<TaskWorkerScheduleInfo> loadTaskWorkerScheduleInfoListByTaskId(String taskId) throws Exception {
        List<TaskWorkerScheduleInfo> list = new ArrayList<TaskWorkerScheduleInfo>();
        if (this.getZooKeeper().exists(getTaskWorkersPath(taskId), false) == null) {
            return list;
        }
        // id/workerIdList
        List<String> workerIdList = this.getZooKeeper().getChildren(getTaskWorkersPath(taskId), false);
        Collections.sort(workerIdList, new Comparator<String>() {//
            public int compare(String leftString, String rightString) {//
                return leftString.substring(leftString.lastIndexOf(ZookeeperNodeConstants.$) + 1)//
                        .compareTo(//
                                rightString.substring(rightString.lastIndexOf(ZookeeperNodeConstants.$) + 1));//
            }
        });

        for (String workerId : workerIdList) {
            list.add(loadTaskWorkerScheduleInfo(taskId, workerId));
        }
        return list;
    }

    public List<TaskWorkerScheduleInfo> loadTaskWorkerScheduleInfoListByWorkerId(String workerId) throws Exception {
        List<TaskWorkerScheduleInfo> list = new ArrayList<TaskWorkerScheduleInfo>();
        List<String> taskIdList = this.getZooKeeper().getChildren(this.taskRootPath, false);
        Collections.sort(taskIdList);
        for (String taskId : taskIdList) {
            String taskIdAndWorkerIdPath = getTaskWorkerPath(taskId, workerId);
            if (this.getZooKeeper().exists(taskIdAndWorkerIdPath, false) != null) {
                list.add(loadTaskWorkerScheduleInfo(taskId, workerId));
            }
        }
        return list;
    }

    public TaskWorkerScheduleInfo loadTaskWorkerScheduleInfo(String taskId, String workerId) throws Exception {
        String taskIdAndWorkerIdPath = getTaskWorkerPath(taskId, workerId);
        if (this.getZooKeeper().exists(taskIdAndWorkerIdPath, false) == null) {
            return null;
        } else {
            byte[] bytes = this.getZooKeeper().getData(taskIdAndWorkerIdPath, false, null);
            if (bytes != null) {
                String value = new String(bytes);
                TaskWorkerScheduleInfo taskWorkerScheduleInfo = JSONObject.parseObject(value, TaskWorkerScheduleInfo.class);
                return taskWorkerScheduleInfo;
                //
            } else {
                TaskWorkerScheduleInfo taskWorkerScheduleInfo = new TaskWorkerScheduleInfo();
                taskWorkerScheduleInfo.setTaskId(taskId);
                taskWorkerScheduleInfo.setWorkerId(workerId);
                taskWorkerScheduleInfo.setRequestNum(-1);//-1表示值不存在
                return taskWorkerScheduleInfo;
            }
        }


    }

    /**
     * @param taskId
     * @param workerId
     * @param requestNum requestNum>0
     * @throws Exception
     */
    public void updateTaskWorkerScheduleInfoRequestNum(String taskId, String workerId, int requestNum) throws Exception {
        String taskIdAndWorkerIdPath = getTaskWorkerPath(taskId, workerId);
        TaskWorkerScheduleInfo taskWorkerScheduleInfo = this.loadTaskWorkerScheduleInfo(taskId, workerId);
        if (taskWorkerScheduleInfo == null) {
            taskWorkerScheduleInfo = new TaskWorkerScheduleInfo();
            taskWorkerScheduleInfo.setTaskId(taskId);
            taskWorkerScheduleInfo.setWorkerId(workerId);
            taskWorkerScheduleInfo.setRequestNum(-1);//-1表示值不存在
        }
        taskWorkerScheduleInfo.setRequestNum(requestNum);//设置请求数
        String value = JSONObject.toJSONString(taskWorkerScheduleInfo);
        this.getZooKeeper().setData(taskIdAndWorkerIdPath, value.getBytes(), -1);
    }

    public List<String> getWorkIdListSupportedForTaskScheduleInfo(String taskId) {
        return null;
    }

    private static class ProcessorComparator implements Comparator<String> { //
        public int compare(String left, String right) {//
            return left.substring(left.lastIndexOf(ZookeeperNodeConstants.$) + 1)//
                    .compareTo(//
                            right.substring(right.lastIndexOf(ZookeeperNodeConstants.$) + 1));//
        }
    }

    private static class ProcessorComparator2 implements Comparator<TaskProcessorInfo> {
        public int compare(TaskProcessorInfo left, TaskProcessorInfo right) {
            int result = left.getTaskId().compareTo(right.getTaskId());//
            if (result == 0) {//
                String leftServerId = left.getProcessorId();//
                String rightServerId = right.getProcessorId();//
                result = leftServerId.substring(leftServerId.lastIndexOf(ZookeeperNodeConstants.$) + 1)//
                        .compareTo(//
                                rightServerId.substring(rightServerId.lastIndexOf(ZookeeperNodeConstants.$) + 1));//
            }
            return result;
        }
    }

    //itemTaskWorker end


}



 